use std::{net::SocketAddr, sync::{atomic::{AtomicU8, Ordering}, Arc}, time::Duration};
use proto::types::{ACK, CONNECT, NOTIFY, NOTIFY_ACK, PING, PONG, REQUEST, RESPONSE, SIMPLE_REQUEST, SIMPLE_RESPONSE};
use tokio::{net::TcpStream, sync::Semaphore, time::timeout};
use crate::{reader::ConnReader, AppErr, ConnectInfo, ErrConv};
use super::{frame::Frame, frame_map::FrameMap, writer::ConnWriter};

#[derive(Debug)]
struct Inner {
    addr: SocketAddr,
    writer: ConnWriter,
    frame_map: FrameMap,    
    read_exit_sem: Semaphore,
    seq: AtomicU8,
    client_id: String,
}

#[derive(Debug, Clone)]
pub struct Client {
    inner: Arc<Inner>,
}

impl Client {

    pub async fn new_wait_connect(conn: TcpStream, addr: SocketAddr) -> Result<(Client, ConnReader), AppErr> {
        let (r, w) = conn.into_split();
        let writer = ConnWriter::new(w);
        let mut reader = ConnReader::new(r);

        let frame = reader.read_with_timeout(Duration::from_secs(5)).await?;
        if frame.ts() != CONNECT {
            return Err(AppErr::Invalid("无效的数据帧"));
        }
        let info: ConnectInfo = frame.parse_connect()?;
       
        let inner = Inner {
            addr,
            writer,
            frame_map: FrameMap::new(),
            read_exit_sem: Semaphore::new(0),
            seq: AtomicU8::new(0),
            client_id: info.client_id.to_owned(),
        };

        let client = Client {
            inner: Arc::new(inner)
        };

        Ok((client, reader))
    }

    pub fn client_id(&self) -> &str {
        &self.inner.client_id
    }

    pub fn addr(&self) -> SocketAddr {
        self.inner.addr
    }

    pub async fn ping(&self) -> Result<(), AppErr> {
        let seq = self.create_seq();
        let id = self.require_id(PONG, seq);
        let rx = self.inner.frame_map.insert(id);
        self.inner.writer.write_min_frame(PING, seq).await?;
        timeout(Duration::from_secs(2), rx).await.map_err(|_| AppErr::Timeout("ping timeout"))?.conv()?;
        Ok(())
    }

    pub async fn request(&self, cmd: u16, body: &[u8], duration: Duration) -> Result<Frame, AppErr> {
        let seq = self.create_seq();
        let ack_rx = self.inner.frame_map.insert(self.require_id(ACK, seq));
        let res_rx = self.inner.frame_map.insert(self.require_id(RESPONSE, seq));
        self.inner.writer.write_frame(REQUEST, seq, cmd, body).await?;
        timeout(Duration::from_secs(2), ack_rx).await.map_err(|_| AppErr::Timeout("ack timeout"))?.conv()?;
        let res_frame = timeout(duration, res_rx).await.map_err(|_| AppErr::Timeout("res timeout"))?.conv()?;
        Ok(res_frame)
    }

    pub async fn simple_request(&self, cmd: u16, body: &[u8]) -> Result<Frame, AppErr> {
        let seq = self.create_seq();
        let res_rx = self.inner.frame_map.insert(self.require_id(SIMPLE_RESPONSE, seq));
        self.inner.writer.write_frame(SIMPLE_REQUEST, seq, cmd, body).await?;
        let frame = timeout(Duration::from_secs(2), res_rx).await.map_err(|_| AppErr::Timeout("simple res timeout"))?.conv()?;
        Ok(frame)
    }

    pub async fn notify(&self, cmd: u16, body: &[u8]) -> Result<(), AppErr> {
        let seq = self.create_seq();
        self.inner.writer.write_frame(NOTIFY, seq, cmd, body).await?;
        Ok(())
    }

    pub async fn notify_ack(&self, cmd: u16, body: &[u8]) -> Result<(), AppErr> {
        let seq = self.create_seq();
        let ack_rx = self.inner.frame_map.insert(self.require_id(ACK, seq));
        self.inner.writer.write_frame(NOTIFY_ACK, seq, cmd, body).await?;
        timeout(Duration::from_secs(2), ack_rx).await.map_err(|_| AppErr::Timeout("ack timeout"))?.conv()?;
        Ok(())
    }

    pub async fn response(&self, seq: u8, cmd: u16, state: u8, body: &[u8]) -> Result<(), AppErr> {
        self.inner.writer.write_res_frame(RESPONSE, seq, cmd, state, body).await?;
        Ok(())
    }

    pub async fn simple_response(&self, seq: u8, cmd: u16, state: u8, body: &[u8]) -> Result<(), AppErr> {
        self.inner.writer.write_res_frame(SIMPLE_RESPONSE, seq, cmd, state, body).await?;
        Ok(())
    }

    pub async fn min_frame(&self, ts: u8, seq: u8) -> Result<(), AppErr> {
        self.inner.writer.write_min_frame(ts, seq).await?;
        Ok(())
    }

    pub fn dispatch(&self, frame: Frame) {
        self.inner.frame_map.dispatch(frame)
    }

    pub async fn close(&self) {
        self.inner.writer.close().await;
        self.inner.read_exit_sem.add_permits(1);
    }

    pub async fn wait_read_exit(&self) {
        _ = self.inner.read_exit_sem.acquire().await;
    }

    fn create_seq(&self) -> u8 {
        let seq = self.inner.seq.fetch_add(1, Ordering::SeqCst);
        seq
    }

    fn require_id(&self, ts: u8, seq: u8) -> u16 {
        ((ts as u16) << 8) +
        (seq as u16)
    }
}














